package com.king.util

import java.util.Properties

import com.king.config.KafkaConfig
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}
import org.apache.kafka.clients.consumer.ConsumerConfig

/**
 * @Author: KingWang
 * @Date: 2022/1/1  
 * @Desc:
 **/
object MyKafkaUtil {


  val broker_list = KafkaConfig.bootstrapServer

  def getKafkaProducer(topic:String):FlinkKafkaProducer[String] =
    new FlinkKafkaProducer[String](broker_list,topic,new SimpleStringSchema())


  def getKafkaConsumer(topic:String,groupId:String):FlinkKafkaConsumer[String] = {
    val prop = new Properties()
    prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker_list)
    prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
    new FlinkKafkaConsumer[String](topic,new SimpleStringSchema(), prop)
  }

}
